#pragma once
#include "../mq_third/include/muduo/proto/codec.h"
#include "../mq_third/include/muduo/proto/dispatcher.h"
#include "../mq_third/include/muduo/base/Logging.h"
#include "../mq_third/include/muduo/base/Mutex.h"
#include "../mq_third/include/muduo/net/EventLoop.h"
#include "../mq_third/include/muduo/net/TcpServer.h"

#include "mq_host.hpp"
#include "mq_connection.hpp"
#include "mq_consumer.hpp"
#include "../mq_comm/mq_threadpool.hpp"
#include "../mq_comm/mq_log.hpp"
#include "../mq_comm/mq_msg.pb.h"
#include "../mq_comm/mq.pb.h"

namespace mq
{
    #define DBFILE "/meta.db"
    #define HOSTNAME "MyVirtualHost"
    class Server
    {
    public:
        using massagePtr = std::shared_ptr<google::protobuf::Message>;
        // onProtobufMessage 根据不同请求查找对应的回调函数
        // ProtobufCodec _codec; ProtobufCodec类内部的 onMessage 会对server收到的数据进行解析, 并调用onProtobufMessage
        // ProtobufCodec对象要传入一个分发器(_dispatcher), 用于解析数据后调用分发器对任务进行分发
        Server(int port, const std::string &basedir) : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port),
            "server", muduo::net::TcpServer::kReusePort),
            _dispatcher(std::bind(&Server::onUnknownMessage, this, std::placeholders::_1, 
                    std::placeholders::_2, std::placeholders::_3)),
            _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
            _virtual_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),
            _consumer_manager(std::make_shared<ConsumerManager>()),
            _connection_manager(std::make_shared<ConnectionManager>()),
            _threadpool(std::make_shared<ThreadPool>())
        {
            //针对历史消息中的所有队列，别忘了，初始化队列的消费者管理结构(每个队列都有一个)
            QueueMap qm = _virtual_host->allQueues();
            for (auto &q : qm) 
            {
                _consumer_manager->initQueueConsumer(q.first);
            }
            _dispatcher.registerMessageCallback<mq::openChannelRequest>(std::bind(&Server::onOpenChannel, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::closeChannelRequest>(std::bind(&Server::onCloseChannel, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::declareExchangeRequest>(std::bind(&Server::onDeclareExchange, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::deleteExchangeRequest>(std::bind(&Server::onDeleteExchange, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::declareQueueRequest>(std::bind(&Server::onDeclareQueue, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::deleteQueueRequest>(std::bind(&Server::onDeleteQueue, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::queueBindRequest>(std::bind(&Server::onQueueBind, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::queueUnBindRequest>(std::bind(&Server::onQueueUnBind, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicPublishRequest>(std::bind(&Server::onBasicPublish, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicAckRequest>(std::bind(&Server::onBasicAck, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicConsumeRequest>(std::bind(&Server::onBasicConsume, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicCancelRequest>(std::bind(&Server::onBasicCancel, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
        }
        void start()
        {
            _server.start();
            _baseloop.loop();
        }

    private:
        // 打开信道
        void onOpenChannel(const muduo::net::TcpConnectionPtr &conn, const openChannelRequestPtr &message, muduo::Timestamp)
        {
            // 根据TcpServer的conn连接, 获取connection对象, 进而打开信道
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "打开信道时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 通过获取到的connection对象来打开信道
            mconn->openChannel(message);
        }
        // 关闭信道
        void onCloseChannel(const muduo::net::TcpConnectionPtr &conn, const closeChannelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "关闭信道时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 通过获取到的connection对象来打开信道
            mconn->closeChannel(message);
        }
        void onDeclareExchange(const muduo::net::TcpConnectionPtr &conn, const declareExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "声明交换机时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 根据connection对象获取到对应的信道
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                LOG(ERROR) << "声明交换机时，没有找到信道！\n";
                return ;
            }

            cp->declareExchange(message);
        }
        // 删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr &conn, const deleteExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "删除交换机时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 根据connection对象获取到对应的信道
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                LOG(ERROR) << "删除交换机时，没有找到信道！\n";
                return ;
            }
            
            cp->deleteExchange(message);
        }
        // 声明队列
        void onDeclareQueue(const muduo::net::TcpConnectionPtr &conn, const declareQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "声明队列时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 根据connection对象获取到对应的信道
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                LOG(ERROR) << "声明队列时，没有找到信道！\n";
                return ;
            }
            
            cp->declareQueue(message);
        }
        // 删除队列
        void onDeleteQueue(const muduo::net::TcpConnectionPtr &conn, const deleteQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "删除队列时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 根据connection对象获取到对应的信道
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                LOG(ERROR) << "删除队列时，没有找到信道！\n";
                return ;
            }    
            cp->deleteQueue(message);
        }
        // 队列绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr &conn, const queueBindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "队列绑定时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 根据connection对象获取到对应的信道
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                LOG(ERROR) << "队列绑定时，没有找到信道！\n";
                return ;
            }
            cp->queueBind(message);
        }
        // 队列解绑
        void onQueueUnBind(const muduo::net::TcpConnectionPtr &conn, const queueUnBindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "队列解绑时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 根据connection对象获取到对应的信道
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                LOG(ERROR) << "队列解绑时，没有找到信道！\n";
                return ;
            }
            cp->queueUnBind(message);
        }
        // 消息发布
        void onBasicPublish(const muduo::net::TcpConnectionPtr &conn, const basicPublishRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                LOG(ERROR) << "消息发布时, 没有找到连接对应的Connection对象!\n";
                conn->shutdown();
                return ;
            }
            // 根据connection对象获取到对应的信道
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                LOG(ERROR) << "消息发布时，没有找到信道！\n";
                return ;
            }
            return cp->basicPublish(message);
        }
        // 消息确认
        void onBasicAck(const muduo::net::TcpConnectionPtr &conn, const basicAckRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                LOG(ERROR) << "确认消息时，没有找到连接对应的Connection对象！\n";
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                LOG(ERROR) << "确认消息时，没有找到信道！\n";
                return;
            }
            return cp->basicAck(message);
        }
        // 队列消息订阅
        void onBasicConsume(const muduo::net::TcpConnectionPtr &conn, const basicConsumeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                LOG(ERROR) << "队列消息订阅时，没有找到连接对应的Connection对象！\n";
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                LOG(ERROR) << "队列消息订阅时，没有找到信道！\n";
                return;
            }
            return cp->basicConsume(message);
        }
        // 队列消息取消订阅
        void onBasicCancel(const muduo::net::TcpConnectionPtr &conn, const basicCancelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                LOG(ERROR) << "队列消息取消订阅时，没有找到连接对应的Connection对象！\n";
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                LOG(ERROR) << "队列消息取消订阅时，没有找到信道！\n";
                return;
            }
            return cp->basicCancel(message);
        }
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const massagePtr& message, muduo::Timestamp) 
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                LOG_INFO << "新连接建立成功！";
                _connection_manager->newConnection(_virtual_host, _consumer_manager, _codec, conn, _threadpool);
            }
            else
            {
                LOG_INFO << "连接即将关闭！";
                _connection_manager->delConnection(conn);
            }
        }

    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        ProtobufDispatcher _dispatcher; //请求分发器对象--要向其中注册请求处理函数
        ProtobufCodecPtr _codec;        // 协议处理

        // 服务管理的一些句柄
        // 传递(构建)一个虚拟机对象, 构建新的Connection对象时, 把虚拟机对象传递过去
        // 到信道部分, 通过虚拟机对象来调用完成消费者对指定的业务
        VirtualHost::ptr _virtual_host;
        // 消费者管理句柄, 主也是传递给Connection对象, Connection再通过这个句柄构建信道
        // 一个消费者管理句柄, 对多个队列的消费者进行管理, 每个消费者都会有与之对应的通信信道
        // 根据消费者的请求(订阅队列, 取消订阅, 声明/删除队列), 进行相应的操作
        ConsumerManager::ptr _consumer_manager;
        // 服务处理: 通过Connection管理句柄, 获取一个连接, 通过连接获取一个信道, 每个信道对应一个消费者, 进行指定业务处理
        ConnectionManager::ptr _connection_manager; 
        ThreadPool::ptr _threadpool; // 一个线程池用来处理一个连接(多个信道)的任务;
    };
}